package com.oner365.test.stream;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * java8 stream 并行流小demo
 */
public class ParallelStreamTestDemo {

  private static final Logger LOGGER = LoggerFactory.getLogger(ParallelStreamTestDemo.class);

  public static void main(String[] args) throws InterruptedException {
    streamForEach();
    LOGGER.info("=================================");
    paralleForEach();
    LOGGER.info("=================================");
    parallelMethod();
  }
  
  public static void streamForEach() {
    long start = System.currentTimeMillis();
    List<Integer> list = new ArrayList<Integer>() {
      private static final long serialVersionUID = 1L;
      {
        add(1);
        add(2);
        add(3);
        add(4);
      }
    };
    list.stream().forEach(i ->{
      sleepForArg(i);
    });
    LOGGER.info("用时:{}", (System.currentTimeMillis() - start) / 1000);
  }

  public static void paralleForEach() {
    long start = System.currentTimeMillis();
    List<Integer> list = new ArrayList<Integer>() {
      private static final long serialVersionUID = 1L;
      {
        add(1);
        add(2);
        add(3);
        add(4);
        add(5);
      }
    };
    list.parallelStream().forEach(i ->{
      sleepForArg(i);
    });
    LOGGER.info("用时:{}", (System.currentTimeMillis() - start) / 1000);
  }

  // 并行逻辑写法
  public static <V> void parallelMethod() throws InterruptedException {
    LOGGER.info("并行数据收集");
    long start = System.currentTimeMillis();
    List<Runnable> list = new ArrayList<Runnable>() {
      private static final long serialVersionUID = 1L;
      {
        add(() -> sleepForArg(1));
        add(() -> sleepForArg(2));
        add(() -> sleepForArg(3));
        add(() -> sleepForArg(4));
      }
    };
    list.parallelStream().forEach(task -> task.run());
    LOGGER.info("用时:{}", (System.currentTimeMillis() - start) / 1000);
  }

  // 模拟大数据量复杂业务逻辑收集数据
  public static void sleepForArg(int time) {
    try {
      TimeUnit.SECONDS.sleep(time);
    } catch (InterruptedException e) {
      LOGGER.error("queryTotalUser error:", e);
    }
    LOGGER.info("执行:sleepForArg time:{}",time);
  }
  
}
